package cn.xiaochao.rpc;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;

import org.apache.avro.AvroRemoteException;
import org.apache.avro.ipc.NettyTransceiver;
import org.apache.avro.ipc.specific.SpecificRequestor;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;

import cn.tedu.zebra.common.GlobalEnv;
import rpc.domain.FileSplit;
import rpc.service.RpcFileSplit;

public class ZkClientConnectRunner implements Runnable {

	private ZooKeeper zk;
	private String path;

	public ZkClientConnectRunner(ZooKeeper zk, String path) {
		super();
		this.zk = zk;
		this.path = path;
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub

		// 1.从切片队列获取Split对象

		try {
			FileSplit fileSplit = GlobalEnv.getSplitQueue().take();

			// 获取节点ip和port

			byte[] data = zk.getData(GlobalEnv.getEngine1path() + "/" + path, null, null);

			String ip = new String(data).split("/")[1];
			int port = Integer.parseInt(new String(data).split("/")[2]);

			// 创建连接rfc服务器

			NettyTransceiver transceiver = new NettyTransceiver(new InetSocketAddress(ip, port));

			final RpcFileSplit rfs = SpecificRequestor.getClient(RpcFileSplit.class, transceiver);

			// 发送切片

			rfs.sendFileSplit(fileSplit);//第一次发
			//永久监听
			while (true) {
				
				final CountDownLatch countDownLatch = new CountDownLatch(1);

				// 监听节点是否繁忙 （第二次需要判断）
				zk.getData(GlobalEnv.getEngine1path() + "/" + path, new Watcher() {

					@Override
					public void process(WatchedEvent we) {
						
						// 监听节点数据改变
						if (we.getType() == EventType.NodeDataChanged) {

							try {
								byte[] nodeDate = zk.getData(GlobalEnv.getEngine1path() + "/" + path, null, null);

								if (new String(nodeDate).endsWith("free")) {

									FileSplit split = GlobalEnv.getSplitQueue().poll();
									
									if (split !=null) {
										rfs.sendFileSplit(split);

									}
									
									countDownLatch.countDown();

								} else {
									
									countDownLatch.countDown();
									
								}

							} catch (KeeperException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							} catch (InterruptedException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							} catch (AvroRemoteException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							}

						}

					}
				}, null);
				
				countDownLatch.await();//释放 只有countDown次数为0才释放
				
			}
			
			

		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

}
